package com.kdpujie.alink;

import java.io.BufferedReader;

import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.model.OSSObject;

import org.apache.flink.api.java.DataSet;
import org.apache.parquet.hadoop.ParquetReader;

import scala.tools.jline_embedded.internal.InputStreamReader;

public class ParquetSource {

    public static void main(String[] args) {
         String schemaStr = "request_id string, ubix_id string, publisher_id string, app_id string, package_name string, app_ver string, app_fix_ver string, province_code string, city_code string, district_code string, os_type string, " +
                        "os_ver string, os_fix_ver string, device_type string, device_brand string, device_model string, language string, hw_model string, hw_machine string, conn_type string, screen_size string, screen_dpi string, screen_orient string, " + 
                        "carrier_type string, carrier_code string, startup_time string, mb_time string, cpu_num string, disk_capacity float, mem_capacity float, auth_status string, battery_power string, request_ipv4 string, request_real_ip string, request_ua string, " + 
                        "ad_slot_id string, ad_type string, ad_is_dpl string, ad_is_macro string, ad_click_region string, creative_id string, creative_type string, creative_title string, creative_desc string, creative_package_name string, creative_img_size string, " + 
                        "creative_video_size string, ads_id string, ads_account_id string, ads_group_id string, ads_campaign_id string, ads_bid_type string, ads_bid_price string, ads_high_ecpm float, ads_second_ecpm float, predict_debug float, " + 
                        "is_dsp_win string, level string, docking_type string, request_hour string, request_weekday string, request_time string, show_ipv4 string, show_ua string, show_time string, show_anti_code string, click_cvu_flag string, click_ipv4 string, click_ua string, " +
                        "click_time string, is_show string, is_click string";

        // yourEndpoint填写Bucket所在地域对应的Endpoint。以华东1（杭州）为例，Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
        String endpoint = "oss-cn-beijing.aliyuncs.com";
        // 阿里云账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维，请登录RAM控制台创建RAM用户。
        String accessKeyId = "LTAI5tHcYbpZoe8JpLn3PnXz";
        String accessKeySecret = "5whjQ8WG0P6h5S8zklhLxVJYhKV352";
        // 填写Bucket名称。
        String bucketName = "ubix-datawarehouse";
        // 填写Object的完整路径。Object完整路径中不能包含Bucket名称。
        String objectName = "user/warehouse/dm/dm_f_algo_model_inc/dt=20210928/part-00000-41f2996a-602c-4b30-98c3-1c16871e0362-c000.snappy.parquet";

        // 创建OSSClient实例。
        OSS ossClient = new OSSClientBuilder().build(endpoint, accessKeyId, accessKeySecret);
        // ossObject包含文件所在的存储空间名称、文件名称、文件元信息以及一个输入流。
        OSSObject ossObject = ossClient.getObject(bucketName, objectName);

        // 读取文件内容。
        System.out.println("Object content:");
        BufferedReader reader = new BufferedReader(new InputStreamReader(ossObject.getObjectContent()));
        
        try {
            while (true) {
                String line = reader.readLine();
                if (line == null) break;
    
                System.out.println("\n" + line);
            }
            // 数据读取完成后，获取的流必须关闭，否则会造成连接泄漏，导致请求无连接可用，程序无法正常工作。
            reader.close();
    
            // 关闭OSSClient。
            ossClient.shutdown();
        } catch (Exception e) {
           e.printStackTrace();
        }
    }
}